package data_propagate.meta;

import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.common.collect.MapMaker;
import com.google.common.collect.Maps;
import com.google.common.collect.MigrateMap;
import data_propagate.meta.exception.DataPropagateMetaManagerException;
import drds.data_propagate.common.AbstractLifeCycle;
import drds.propagate.protocol.ClientId;
import drds.propagate.protocol.position.Position;
import drds.propagate.protocol.position.PositionRange;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 内存版实现
 *
 * @author zebin.xuzb @ 2012-7-2
 * @version 1.0.0
 */
public class MemoryMetaManager extends AbstractLifeCycle implements MetaManager {

    protected Map<String, List<ClientId>> destinations;
    protected Map<ClientId, MemoryClientIdentityBatch> batches;
    protected Map<ClientId, Position> cursors;

    public void start() {
        super.start();

        batches = MigrateMap.makeComputingMap(new Function<ClientId, MemoryClientIdentityBatch>() {

            public MemoryClientIdentityBatch apply(ClientId clientId) {
                return MemoryClientIdentityBatch.create(clientId);
            }

        });

        cursors = new MapMaker().makeMap();

        destinations = MigrateMap.makeComputingMap(new Function<String, List<ClientId>>() {

            public List<ClientId> apply(String destination) {
                return Lists.newArrayList();
            }
        });
    }

    public void stop() {
        super.stop();

        destinations.clear();
        cursors.clear();
        for (MemoryClientIdentityBatch batch : batches.values()) {
            batch.clearPositionRanges();
        }
    }

    public synchronized void subscribe(ClientId clientId) throws DataPropagateMetaManagerException {
        List<ClientId> clientIds = destinations.get(clientId.getDestination());

        if (clientIds.contains(clientId)) {
            clientIds.remove(clientId);
        }

        clientIds.add(clientId);
    }

    public synchronized boolean hasSubscribe(ClientId clientId) throws DataPropagateMetaManagerException {
        List<ClientId> clientIds = destinations.get(clientId.getDestination());
        return clientIds != null && clientIds.contains(clientId);
    }

    public synchronized void unsubscribe(ClientId clientId) throws DataPropagateMetaManagerException {
        List<ClientId> clientIds = destinations.get(clientId.getDestination());
        if (clientIds != null && clientIds.contains(clientId)) {
            clientIds.remove(clientId);
        }
    }

    public synchronized List<ClientId> listAllSubscribeInfo(String destination) throws DataPropagateMetaManagerException {
        // fixed issue #657, fixed ConcurrentModificationException
        return Lists.newArrayList(destinations.get(destination));
    }

    public Position getCursor(ClientId clientId) throws DataPropagateMetaManagerException {
        return cursors.get(clientId);
    }

    public void updateCursor(ClientId clientId, Position position) throws DataPropagateMetaManagerException {
        cursors.put(clientId, position);
    }

    public Long addBatch(ClientId clientId, PositionRange positionRange) throws DataPropagateMetaManagerException {
        return batches.get(clientId).addPositionRange(positionRange);
    }

    public void addBatch(ClientId clientId, PositionRange positionRange, Long batchId)
            throws DataPropagateMetaManagerException {
        batches.get(clientId).addPositionRange(positionRange, batchId);// 添加记录到指定batchId
    }

    public PositionRange removeBatch(ClientId clientId, Long batchId) throws DataPropagateMetaManagerException {
        return batches.get(clientId).removePositionRange(batchId);
    }

    public PositionRange getBatch(ClientId clientId, Long batchId) throws DataPropagateMetaManagerException {
        return batches.get(clientId).getPositionRange(batchId);
    }

    public PositionRange getLastestBatch(ClientId clientId) throws DataPropagateMetaManagerException {
        return batches.get(clientId).getLastestPositionRange();
    }

    public PositionRange getFirstBatch(ClientId clientId) throws DataPropagateMetaManagerException {
        return batches.get(clientId).getFirstPositionRange();
    }

    public Map<Long, PositionRange> listAllBatchs(ClientId clientId) throws DataPropagateMetaManagerException {
        return batches.get(clientId).listAllPositionRange();
    }

    public void clearAllBatchs(ClientId clientId) throws DataPropagateMetaManagerException {
        batches.get(clientId).clearPositionRanges();
    }

    // ============================

    public static class MemoryClientIdentityBatch {

        private ClientId clientId;
        private Map<Long, PositionRange> batches = new MapMaker().makeMap();
        private AtomicLong atomicMaxBatchId = new AtomicLong(1);

        public MemoryClientIdentityBatch() {

        }

        protected MemoryClientIdentityBatch(ClientId clientId) {
            this.clientId = clientId;
        }

        public static MemoryClientIdentityBatch create(ClientId clientId) {
            return new MemoryClientIdentityBatch(clientId);
        }

        public synchronized void addPositionRange(PositionRange positionRange, Long batchId) {
            updateMaxId(batchId);
            batches.put(batchId, positionRange);
        }

        public synchronized Long addPositionRange(PositionRange positionRange) {
            Long batchId = atomicMaxBatchId.getAndIncrement();
            batches.put(batchId, positionRange);
            return batchId;
        }

        public synchronized PositionRange removePositionRange(Long batchId) {
            if (batches.containsKey(batchId)) {
                Long minBatchId = Collections.min(batches.keySet());
                if (!minBatchId.equals(batchId)) {
                    // 检查一下提交的ack/rollback，必须按batchId分出去的顺序提交，否则容易出现丢数据
                    throw new DataPropagateMetaManagerException(String.format("batchId:%d is not the firstly:%d",
                            batchId,
                            minBatchId));
                }
                return batches.remove(batchId);
            } else {
                return null;
            }
        }

        public synchronized PositionRange getPositionRange(Long batchId) {
            return batches.get(batchId);
        }

        public synchronized PositionRange getLastestPositionRange() {
            if (batches.size() == 0) {
                return null;
            } else {
                Long batchId = Collections.max(batches.keySet());
                return batches.get(batchId);
            }
        }

        public synchronized PositionRange getFirstPositionRange() {
            if (batches.size() == 0) {
                return null;
            } else {
                Long batchId = Collections.min(batches.keySet());
                return batches.get(batchId);
            }
        }

        public synchronized Map<Long, PositionRange> listAllPositionRange() {
            Set<Long> batchIdSets = batches.keySet();
            List<Long> batchIds = Lists.newArrayList(batchIdSets);
            Collections.sort(Lists.newArrayList(batchIds));

            return Maps.newHashMap(batches);
        }

        public synchronized void clearPositionRanges() {
            batches.clear();
        }

        private synchronized void updateMaxId(Long batchId) {
            if (atomicMaxBatchId.get() < batchId + 1) {
                atomicMaxBatchId.set(batchId + 1);
            }
        }

        // ============ setter & getter =========

        public ClientId getClientId() {
            return clientId;
        }

        public void setClientId(ClientId clientId) {
            this.clientId = clientId;
        }

    }

}
